package com.lost.octopus.es.processor.mq.consumer.capture;

import com.google.common.collect.Lists;
import com.lost.octopus.es.processor.mq.consumer.dto.SinkDeal;
import com.lost.octopus.es.processor.scheduler.EsSyncIndexOperator;
import com.lost.octopus.es.processor.scheduler.EsSyncManagement;
import com.lost.octopus.es.processor.scheduler.entity.BurstResult;
import lombok.extern.log4j.Log4j2;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 数据捕获-分片处理管理器
 * @author zhangbh
 * @date 2025/5/10 11:48
 */
@Log4j2
public class BurstCaptureManager {

    /**
     * 同步助手
     */
    @Resource
    EsSyncManagement esSyncManagement;

    void burstDataCapture(List<String> allIds, Class<?> idType, SinkDeal info) {
        // 分片数量
        int burstSize = 1000;
        EsSyncIndexOperator<?, ?, ?, ?> opt = info.getOpt();
        log.debug("desc {} idType {} 查询数据={} 准备检查同步=========", info.getDesc(), idType.getSimpleName(), allIds);
        if (idType == String.class) {
            List<List<String>> part = Lists.partition(allIds, burstSize);
            EsSyncIndexOperator<?, String, ?, ?> strOpt = (EsSyncIndexOperator<?, String, ?, ?>) opt;
            // 数据是esId 从数据库查询到数据就更新，没有直接删除
            // 目前简单做，直接删除再查询
            List<String> effectIds = new ArrayList<>();
            for (List<String> partitionIds : part) {
                IndexCoordinates indexCoordinates = info.getOpt().getEsRepository().getIndexCoordinates();
                // 需要支持全量
                BurstResult<String> result = esSyncManagement.saveOrDelEsByEsId(partitionIds, strOpt, indexCoordinates);
                effectIds.addAll(result.getIds());
            }
            log.debug("esId desc {} 有效数据={}  {}=========", info.getDesc(), effectIds.size(), effectIds);
        } else {
            // 数据是主表id 同步处理
            List<Integer> ids = allIds.stream().map(value -> changeEsValue(value, info)).collect(Collectors.toList());
            List<List<Integer>> part = Lists.partition(ids, burstSize);
            // 目前简单做，直接删除再查询
            List<Object> effectIds = new ArrayList<>();
            for (List<Integer> partitionIds : part) {
                IndexCoordinates indexCoordinates = info.getOpt().getEsRepository().getIndexCoordinates();
                // 需要支持全量
                List<?> list = esSyncManagement.saveOrDelForEs(partitionIds, info.getOpt(), indexCoordinates, false);
                effectIds.addAll(list);
            }
            log.debug("id desc {} 有效数据={}  {}=========", info.getDesc(), effectIds.size(), effectIds);
        }
    }

    private Integer changeEsValue(String id, SinkDeal info) {
        try {
            return Integer.valueOf(id);
        } catch (Exception e) {
            throw new IllegalArgumentException(String.format("请确认关联sql注册正确， 操作 %s,对应的表%s变更，需要查询到int型id",
                info.getDesc(), info.getOpt().businessKey().getLabel()));
        }
    }

}
